package com.deng.eventflow.core;

import com.deng.eventflow.core.constant.NextStatus;
import com.deng.eventflow.core.exception.ProcessorException;
import com.deng.eventflow.core.exception.PullableException;
import com.deng.eventflow.core.exception.PushableException;

import java.util.ArrayDeque;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.Future;

public abstract class SynchronousProcessor extends Processor {

    protected final transient Queue<Object[]> m_tempQueue;

    protected final transient Pushable[] m_inputPushables;

    protected transient Pullable[] m_outputPullables;

    public SynchronousProcessor(int in_arity, int out_arity) {
        super(in_arity, out_arity);
        m_tempQueue = new ArrayDeque<Object[]>(1);
        m_inputPushables = new Pushable[in_arity];
        m_outputPullables = new Pullable[out_arity];
    }


    @Override
    public synchronized Pushable getPushableInput(int index) {
        if (m_inputPushables[index] == null) {
            m_inputPushables[index] = new InputPushable(index);
        }
        return m_inputPushables[index];

    }

    @Override
    public Pullable getPullableOutput(int index) {
        if (m_outputPullables[index] == null) {
            m_outputPullables[index] = new OutputPullable(index);
        }
        return m_outputPullables[index];

    }

    protected abstract boolean compute(Object[] inputs, Queue<Object[]> outputs);

    protected class InputPushable implements Pushable
    {
        /**
         * The index of the processor's input this pushable refers to
         */
        private final int m_index;

        /**
         * Creates a pushable associated to some of a processor's input traces.
         *
         * @param index
         *          The index of the trace. Should be between 0 and the processor's
         *          input arity - 1. This is not checked by the constructor, so beware.
         */
        InputPushable(int index)
        {
            super();
            m_index = index;
        }

        @Override
        public final synchronized Future<Pushable> pushFast(Object o)
        {
            push(o);
            return Pushable.NULL_FUTURE;
        }

        @Override
        public synchronized int getPosition()
        {
            return m_index;
        }

        @Override
        public synchronized Pushable push(Object o)
        {
            try
            {
                Queue<Object> q = m_inputQueues[m_index];
                q.add(o);
            }
            catch (ArrayIndexOutOfBoundsException e)
            {
                throw new PushableException(e);
            }
            // Check if each input queue has an event ready
            for (int i = 0; i < m_inputArity; i++)
            {
                Queue<Object> queue = m_inputQueues[i];
                if (queue.isEmpty())
                {
                    // One of them doesn't: we can't produce an output yet
                    return this;
                }
            }
            // Pick an event from each input queue
            Object[] inputs = new Object[m_inputArity];
            for (int i = 0; i < m_inputArity; i++)
            {
                Queue<Object> queue = m_inputQueues[i];
                Object ob = queue.remove();
                inputs[i] = ob;
            }
            // Compute output event
            m_tempQueue.clear();
            boolean outs;
            try
            {
                outs = compute(inputs, m_tempQueue);
            }
            catch (ProcessorException e)
            {
                throw new PushableException(e);
            }
            // put everything in outputEvent, because of duplicated code in notifyEndOfTrace
            outputEvent(outs);
            return this;
        }

        @Override
        public void notifyEndOfTrace()
        {
            m_hasBeenNotifiedOfEndOfTrace[m_index] = true;
            if (!allNotifiedEndOfTrace())
            {
                return;
            }
            m_tempQueue.clear();
            boolean outs;
            try
            {
                outs = onEndOfTrace(m_tempQueue);
            }
            catch (ProcessorException e)
            {
                throw new PushableException(e);
            }
            outputEvent(outs);

            // Notifies the output pushables of the end of the trace
            for (int i = 0; i < m_outputPushables.length; i++)
            {
                Pushable p = m_outputPushables[i];
                if (p == null)
                {
                    throw new PushableException("Output " + i
                            + " of this processor is connected to nothing", getProcessor());
                }
                p.notifyEndOfTrace();
            }
        }

        /**
         * Pushes output event (if any) to the corresponding output {@link Pushable}s.
         *
         * @param outs Set to <tt>true</tt> to enable the output of an event,
         * <tt>false</tt> otherwise.
         */
        private final void outputEvent(boolean outs)
        {
            if (outs && !m_tempQueue.isEmpty())
            {
                for (Object[] evt : m_tempQueue)
                {
                    if (evt != null)
                    {
                        for (int i = 0; i < m_outputPushables.length; i++)
                        {
                            Pushable p = m_outputPushables[i];
                            if (p == null)
                            {
                                throw new PushableException(
                                        "Output " + i + " of processor " + SynchronousProcessor.this
                                                + " is connected to nothing", getProcessor());
                            }
                            p.push(evt[i]);
                        }
                    }
                }
                m_tempQueue.clear();
            }
        }

        @Override
        public synchronized Processor getProcessor()
        {
            return SynchronousProcessor.this;
        }
    }

    protected class OutputPullable implements Pullable
    {
        /**
         * The index of the processor's output this pullable refers to
         */
        private final int m_index;

        /**
         * Creates a pullable associated to some of a processor's output traces.
         *
         * @param index
         *          The index of the trace. Should be between 0 and the processor's
         *          output arity - 1. This is not checked by the constructor, so beware.
         */
        public OutputPullable(int index)
        {
            super();
            m_index = index;
        }

        @Override
        public synchronized void remove()
        {
            // Cannot remove an event on a pullable
            throw new UnsupportedOperationException();
        }

        @Override
        public synchronized Object pullSoft()
        {
            if (hasNextSoft() != NextStatus.YES)
            {
                return null;
            }
            synchronized (m_outputQueues)
            {
                Queue<Object> out_queue = m_outputQueues[m_index];
                // If an event is already waiting in the output queue,
                // return it and don't pull anything from the input
                if (!out_queue.isEmpty())
                {
                    return out_queue.remove();
                }
            }
            return null;
        }

        @Override
        public synchronized Object pull()
        {
            if (!hasNext())
            {
                throw new NoSuchElementException();
            }
            synchronized (m_outputQueues)
            {
                Queue<Object> out_queue = m_outputQueues[m_index];
                // If an event is already waiting in the output queue,
                // return it and don't pull anything from the input
                if (!out_queue.isEmpty())
                {
                    return out_queue.remove();
                }
            }
            throw new NoSuchElementException();
        }

        @Override
        @SuppressWarnings("squid:S2272") // since() pull throws the exception
        public final synchronized Object next()
        {
            return pull();
        }

        @Override
        public synchronized boolean hasNext()
        {
            Queue<Object> out_queue = m_outputQueues[m_index];
            // If an event is already waiting in the output queue,
            // return it and don't pull anything from the input
            if (!out_queue.isEmpty())
            {
                return true;
            }
            // Check if each pullable has an event ready
            for (int tries = 0; tries < MAX_PULL_RETRIES; tries++)
            {
                for (int i = 0; i < m_inputArity; i++)
                {
                    Pullable p = m_inputPullables[i];
                    if (p == null)
                    {
                        throw new PullableException("Input " + i + " of processor " + SynchronousProcessor.this
                                + " is connected to nothing", getProcessor());
                    }
                    boolean status = p.hasNext();
                    if (!status)
                    {
                        if (allNotifiedEndOfTrace())
                        {
                            return false;
                        }
                        Queue<Object[]> last_queue = new ArrayDeque<Object[]>();
                        boolean b = onEndOfTrace(last_queue);
                        m_hasBeenNotifiedOfEndOfTrace[i] = true;
                        if (!b)
                        {
                            return false;
                        }
                        for (Object[] front : last_queue)
                        {
                            for (int j = 0; j < m_outputArity; j++)
                            {
                                m_outputQueues[j].add(front[j]);
                            }
                        }
                        return true;
                    }
                }
                // We are here only if every input pullable has answered YES
                // Pull an event from each
                Object[] inputs = new Object[m_inputArity];
                for (int i = 0; i < m_inputArity; i++)
                {
                    Pullable p = m_inputPullables[i];
                    // Don't check for p == null, we did it above
                    Object o = p.pull();
                    inputs[i] = o;
                }
                // Compute output event(s)
                m_tempQueue.clear();
                boolean computed;
                try
                {
                    computed = compute(inputs, m_tempQueue);
                }
                catch (ProcessorException e)
                {
                    throw new PullableException(e);
                }
                NextStatus status_to_return = NextStatus.NO;
                if (!computed)
                {
                    // No output will ever be returned: stop there
                    return false;
                }
                if (!m_tempQueue.isEmpty())
                {
                    // We computed an output event; add it to the output queue
                    // and answer YES
                    for (Object[] evt : m_tempQueue)
                    {
                        if (evt != null)
                        {
                            for (int i = 0; i < m_outputArity; i++)
                            {
                                Queue<Object> queue = m_outputQueues[i];
                                queue.add(evt[i]);
                            }
                            status_to_return = NextStatus.YES;
                        }
                        else
                        {
                            // This source will NEVER output anything again
                            m_tempQueue.clear();
                            return false;
                        }
                    }
                    if (status_to_return == NextStatus.YES)
                    {
                        m_tempQueue.clear();
                        return true;
                    }
                }
                // Otherwise, try the whole thing again
            }
            m_tempQueue.clear();
            return false;
        }

        @Override
        public synchronized NextStatus hasNextSoft()
        {
            Queue<Object> out_queue = m_outputQueues[m_index];
            // If an event is already waiting in the output queue,
            // return yes and don't pull anything from the input
            if (!out_queue.isEmpty())
            {
                return NextStatus.YES;
            }
            // Check if each pullable has an event ready
            for (int i = 0; i < m_inputArity; i++)
            {
                Pullable p = m_inputPullables[i];
                NextStatus status = p.hasNextSoft();
                if (status == NextStatus.NO)
                {
                    if (allNotifiedEndOfTrace())
                    {
                        return NextStatus.NO;
                    }
                    Queue<Object[]> last_queue = new ArrayDeque<Object[]>();
                    boolean b = onEndOfTrace(last_queue);
                    m_hasBeenNotifiedOfEndOfTrace[i] = true;
                    if (!b)
                    {
                        return NextStatus.NO;
                    }
                    for (Object[] front : last_queue)
                    {
                        for (int j = 0; j < m_outputArity; j++)
                        {
                            m_outputQueues[j].add(front[j]);
                        }
                    }
                    return NextStatus.YES;
                }
                if (status == NextStatus.MAYBE)
                {
                    return NextStatus.MAYBE;
                }
            }
            // We are here only if every input pullable has answered YES
            // Pull an event from each
            Object[] inputs = new Object[m_inputArity];
            {
                int i = 0;
                for (Pullable p : m_inputPullables)
                {
                    inputs[i] = p.pullSoft();
                    i++;
                }
            }
            // Compute output event(s)
            NextStatus status_to_return = NextStatus.MAYBE;
            m_tempQueue.clear();
            boolean computed;
            try
            {
                computed = compute(inputs, m_tempQueue);
            }
            catch (ProcessorException e)
            {
                throw new PullableException(e);
            }
            if (!computed)
            {
                status_to_return = NextStatus.NO;
            }
            else if (!m_tempQueue.isEmpty())
            {
                for (Object[] evt : m_tempQueue)
                {
                    if (evt != null)
                    {
                        // We computed an output event; add it to the output queue
                        // and answer YES
                        int i = 0;
                        for (Queue<Object> queue : m_outputQueues)
                        {
                            queue.add(evt[i]);
                            i++;
                        }
                        status_to_return = NextStatus.YES;
                    }
                }
            }
            return status_to_return;
        }

        @Override
        public synchronized Processor getProcessor()
        {
            return SynchronousProcessor.this;
        }

        @Override
        public synchronized int getPosition()
        {
            return m_index;
        }

        @Override
        public synchronized Iterator<Object> iterator()
        {
            return this;
        }

        @Override
        public synchronized void start()
        {
            // Do nothing
        }

        @Override
        public synchronized void stop()
        {
            // Do nothing
        }

        @Override
        public synchronized void dispose()
        {
            // Do nothing
        }
    }

    @Override
    public final SynchronousProcessor setEventTracker(EventTracker tracker) {
        super.setEventTracker(tracker);
        return this;
    }
}
